package org.swift.mashup.engine.concurrency.memory;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.swift.commons.xml.Match;
import org.w3c.dom.Node;

import com.eviware.soapui.support.xml.XmlUtils;

/**
 * flush pushes the items from the memory buffer into the working memory
**/
@SuppressWarnings("unchecked")
public class WorkingMemory {

	private static final Logger log = Logger.getLogger(WorkingMemory.class);

	private MemoryList currentMemory = new MemoryList();
	private MemoryList limboBuffer = new MemoryList();
	private MemoryList memoryBuffer = new MemoryList();
	private Map<String, List> itemGroups = new HashMap<String, List>();
	private Map<String, MemoryType> memoryTypes = new HashMap<String, MemoryType>();
	private List<MemoryEventListener> listeners = new ArrayList<MemoryEventListener>();
	private int itemCount = 0;
	private Iterator iterator;
	private boolean persistent = false;
	private String workingDirectory;
	private String memoryID;
	private String processID;
	private MappedMemoryManager writeableMemory;
	private Set<String> maybeCollectedTypes = new HashSet<String>();
	private Map<String, MappedMemoryManager> readOnlyMemory = new HashMap<String, MappedMemoryManager>();
	
	public WorkingMemory(Map<String, MemoryType> memoryTypes){
		addMemoryTypes(memoryTypes);
	}
	public WorkingMemory(Map<String, MemoryType> memoryTypes, String workingDirectory, String memoryID, String processID,boolean persistent){
		this.memoryID = memoryID;
		this.workingDirectory = workingDirectory;
		this.processID = processID;
		addMemoryTypes(memoryTypes);
		this.setPersistent(persistent);
	}
	public WorkingMemory(Map<String, MemoryType> memoryTypes, String workingDirectory, String memoryID, String processID, int limit, int newMemoryPosition){
		this.memoryID = memoryID;
		this.workingDirectory = workingDirectory;
		this.processID = processID;
		addMemoryTypes(memoryTypes);
		createMemoryFiles(workingDirectory, memoryID, processID, limit, newMemoryPosition);
		this.persistent = true;
	}
	
	private void addMemoryTypes(Map<String, MemoryType> memoryTypes){
		for (MemoryType memoryType: memoryTypes.values()){
			addMemoryItemGroup(memoryType);
		}
		this.memoryTypes = memoryTypes;
	}
	
	public Map<String, MemoryType> getMemoryTypes() {
		return memoryTypes;
	}
	
	public String getWorkingDirectory() {
		return workingDirectory;
	}
	public void setWorkingDirectory(String workingDirectory) {
		this.workingDirectory = workingDirectory;
	}
	public Set<MemoryType> getBufferedMemoryTypes(){
		Set<MemoryType> bufTypes = new HashSet<MemoryType>();
		for (MemoryItem item:this.memoryBuffer) {
			bufTypes.add(item.type);
		}
		return bufTypes;
	}
	private void addMemoryItemGroup(MemoryType type){
		List<MemoryItem> itemGroup = new ArrayList<MemoryItem>();
		this.itemGroups.put(type.getName(), itemGroup);		
	}
	private void createMemoryFiles(String workingDirectory, String memoryID, String processID, int limit, int newMemoryPosition) {
		File outputDir = new File(workingDirectory);
		outputDir.mkdirs();
		File inputMapFile;
		//Open input files
		inputMapFile = new File(outputDir, "mm-" + memoryID + ".map");
		/*
		 * If the file referenced by the memoryID and processId
		 * exists it is loaded and added to the read-only managers.
		 */
		if (inputMapFile.exists()){
			MappedMemoryManager memManager = new MappedMemoryManager();
			memManager.open(workingDirectory, memoryID, true);
			this.readOnlyMemory.put("0", memManager);
			MappedMemoryManager.loadMemory(this, memoryID, "0", newMemoryPosition, limit, false);
		}
		this.writeableMemory = new MappedMemoryManager();
		if (processID!=null){
			this.writeableMemory.setId(processID);
			this.writeableMemory.open(workingDirectory, memoryID + "-" + processID, false);
		} else {
			this.writeableMemory.open(workingDirectory, memoryID, false);
		}
	}
	public void insert(String type, Object item) {
		if (this.memoryTypes.containsKey(type)){
			this.addToBuffer(type, item);
		} else {
			addMemoryType(type);
			this.addToBuffer(type, item);
		}
	}
	private void addMemoryType(String type){
		if ( !type.equalsIgnoreCase("ERROR") ) {
			Exception howDidWeGetHere = new RuntimeException("ALERT THE AUTHORITIES!");
			howDidWeGetHere.fillInStackTrace();
			log.warn("DYNAMICALLY CREATING NEW MEMORY TYPE '" + type + "'", howDidWeGetHere);
		}

		MemoryType mt = new MemoryType(this.memoryTypes.size()+1, type);
		this.memoryTypes.put(mt.getName(), mt);
		addMemoryItemGroup(mt);
	}
	private void addToBuffer(String type, Object item){
		if(StringUtils.isEmpty(type)) {
			throw new IllegalArgumentException("Memory type must not be null or empty");
		}
		if(item == null) {
			throw new IllegalArgumentException("Memory item must not be null");
		}
		MemoryType memType = this.memoryTypes.get(type);
		MemoryItem memItem;
		if (item!=null && MemoryItem.class.isAssignableFrom(item.getClass())){
			memItem = (MemoryItem)item;
			memItem.id = ++itemCount;
		} else {
			memItem = WorkingMemory.newMemoryItem(++itemCount,item,memType,this,this.writeableMemory);
		}
		this.memoryBuffer.add(memItem);
		for (MemoryEventListener listener:listeners){
			MemoryEvent event = new MemoryEvent(memItem,this,MemoryEvent.eventType.AddedToBuffer);
			listener.handleEvent(event);	
		}		
	}
	
/**
 * addToWorkingMemory is call when moving the memory item from the buffer
 * to current memory.
 * It delegates the work to addItemToWorkingMemory so that there is a version
 * of the method that fires the events and a version that does not.
 * When executing in a parallel mode the events are suppressed.
 * 
 * @param item
 */
	private void addToWorkingMemory(MemoryItem item){
		emptyLimboMemory(item.id);
		addItemToWorkingMemory(item);
		for (MemoryEventListener listener:listeners){
			MemoryEvent event = new MemoryEvent(item,this,MemoryEvent.eventType.AddedToWorkingMemory);
			listener.handleEvent(event);	
		}
	}
	private void emptyLimboMemory(int id){
		if (this.limboBuffer.size()==0) return;
		List limboItemToRemove = new ArrayList();
		for (MemoryItem item:this.limboBuffer){
			if (item.id < id) {
				addItemToWorkingMemory(item);
				limboItemToRemove.add(item);
			}
		}
		this.limboBuffer.removeAll(limboItemToRemove);
	}
/**
 * addItemToWorkingMemory is call when moving the memory item from the buffer 
 * to current memory.
 * When executing in a parallel mode it is called because the events 
 * are suppressed (not raised).
 * 
 * @param item
 */
	protected void addItemToWorkingMemory(MemoryItem item){
		this.currentMemory.add(item);
		String type = item.getType().getName();
		if (this.itemGroups.containsKey(type)){
			List itemGroupList = this.itemGroups.get(type);
			itemGroupList.add(item);
		}
	}
	
/**
 * addItemToWorkingMemoryBuffer is a method used when load memory items from an
 * existing file into the buffer to be processed but without raising the buffer events.
 * This method is called by sub-processes running in parallel mode.
 * 
 * @param item
 */
	protected void addItemToWorkingMemoryBuffer(MemoryItem item){
		if(item == null) {
			throw new IllegalArgumentException("Memory item must not be null");
		}
		this.memoryBuffer.add(item);
	}
	protected void addItemToLimbo(MemoryItem item){
		this.limboBuffer.add(item);
	}
	public void clear() {
		this.currentMemory.clear();
		this.memoryBuffer.clear();
		for (List itemGroup: this.itemGroups.values()){
			itemGroup.clear();
		}
		destroyMemoryFiles();
	}
	
	public void insertAll(Map<String, Object> items) {
		for(Entry entry: items.entrySet()){
			this.insert((String)entry.getKey(), entry.getValue());
		}
	}
	
	
	public MappedMemoryManager getWriteableMemory() {
		return writeableMemory;
	}
	public Map<String, MappedMemoryManager> getReadOnlyMemory() {
		return readOnlyMemory;
	}

	public static MemoryItem newMemoryItem(int id, Object item, MemoryType type, WorkingMemory workingMemory, MappedMemoryManager memoryManager) {
		if (workingMemory.isPersistent()){

			if (type.getName().startsWith("collect(")){
				return new CollectionMemoryItem(id,type,workingMemory);
			} else if (item instanceof Node) {
				String value = XmlUtils.serialize((Node)item, true);
				return new PersistentMemoryItem(id,value,type,workingMemory);
			
			} else if (item instanceof InputStream) {
				//TODO hold the stream till requested
				try {
					byte[] bytes = IOUtils.toByteArray((InputStream)item);
					return new PersistentMemoryItem(id,bytes,type,workingMemory);
				} catch (IOException e) {
					throw new RuntimeException(e);
				}
			} else if (item instanceof Exception) {
				return new ExceptionMemoryItem(id, (Exception)item, type);
			
			} else {
				return new PersistentMemoryItem(id,item,type,workingMemory);
			}
		} else {
			if (item instanceof Node) {
				return new W3CMemoryItem(id, (Node)item, type);
			} else if (item instanceof Collection) {
				return new CollectionMemoryItem(id, (Collection)item, type);
			} else if (item instanceof String) {
				return new BytesMemoryItem(id, (String)item, type);
			} else if (item instanceof byte[]) {
				return new BytesMemoryItem(id, (byte[])item, type);
			} else if (item instanceof InputStream) {
				//TODO hold the stream till requested
				try {
					byte[] bytes = IOUtils.toByteArray((InputStream)item);
					return new BytesMemoryItem(id, bytes, type);
				} catch (IOException e) {
					throw new RuntimeException(e);
				}
			} else if (item instanceof Exception) {
				return new ExceptionMemoryItem(id, (Exception)item, type);				
			} else if (item instanceof Match) {
				//TODO probably shouldn't read this here
				return new BytesMemoryItem(id, ((Match)item).getValue(), type);
			} else if (item instanceof Number) {
				return new NumericMemoryItem(id, (Number)item, type);
			} else if (item instanceof Boolean) {
				return new BytesMemoryItem(id, item.toString(), type);
			} else {
				if (item != null) {			
					throw new RuntimeException("Could not create a memory item for item type '" + item.getClass() + "', memory type '" + type.getName() + "'");
				} else {
					throw new NullPointerException("Could not crete a memory item from null");
				}
			}
		}
	}
	public static MemoryItem newCollectionMemoryItem(int id, MemoryType type, WorkingMemory workingMemory) {
			return new CollectionMemoryItem(id, type, workingMemory);
	}
	
	public void flush() {
		flush(false);
	}
	
/**
 * The suppressEvents parameter is used when running in parallel mode to
 * allow the accumulation of memory item add events.  These accumulated
 * events are handed to the sub-process to execute.
 * When suppressEvents is true the addToWorkingMemory event is not raised
 * for each memory item.  Only the flush event is raised when all items
 * have been added.
 * 	
 * @param suppressEvents
 */
	public void flush(boolean suppressEvents) {
		int startId = Integer.MAX_VALUE;
		int endId = 0;
		MemoryItem lastItem = null;
		for (MemoryItem item:this.memoryBuffer){
			startId = Math.min(startId, item.id);
			endId = Math.max(endId, item.id);
			if (suppressEvents) addItemToWorkingMemory(item); else addToWorkingMemory(item); 
			maybeCollectedTypes.add(item.getType().getName());
			lastItem = item;
		}
		this.memoryBuffer.clear();
		for (MemoryEventListener listener:listeners){
			MemoryEvent event = new MemoryEvent(lastItem,this,MemoryEvent.eventType.MemoryFlushed);
			event.setStartId(startId);
			event.setEndId(endId);
			listener.handleEvent(event);	
		}
	}

	public void addCollectsToBuffer(boolean suppressEvents) {
		//Defer the adding of collections until all sub-processes have completed
		if (!suppressEvents && this.processID==null) {
			for (String type: maybeCollectedTypes){
				MemoryType memType = this.memoryTypes.get(type);
				if (memType.isCollect()) {
					MemoryType collectType = this.memoryTypes.get("collect(" + type + ")");
					MemoryItem memoryItem = WorkingMemory.newCollectionMemoryItem(++itemCount,collectType,this);
					addItemToWorkingMemoryBuffer(memoryItem); 
				}
			}
			maybeCollectedTypes.clear();
		}
	}

	public void addListener(MemoryEventListener listener) {
		this.listeners.add(listener);
	}
	
	public void removeListener(MemoryEventListener listener) {
		this.listeners.remove(listener);	
	}

	public BitSet getWorkingMemoryMask() {
		return currentMemory.getMask();
	}
	public boolean hasNewMemoryItems(){
		return (memoryBuffer.size()>0);
	}

	public Iterator getIterator() {
		return iterator;
	}
		
	public List<MemoryItem> getItems(String memoryType){
			if (this.itemGroups.containsKey(memoryType)){
				return this.itemGroups.get(memoryType);				
			}
			return new ArrayList<MemoryItem>();
	}
	protected boolean isPersistent() {
		return persistent;
	}
	protected void setPersistent(boolean persistent) {
		if (this.persistent!=persistent){
			if(persistent){
				createMemoryFiles(workingDirectory, memoryID, processID, -1, -1);
				persistExistingMemoryItems();
			} else {
				populatePersistedMemoryItems();
				destroyMemoryFiles();
			}		
		}
		this.persistent = persistent;
	}
	public void destroyMemoryFiles() {
		if(this.writeableMemory != null) {
			this.writeableMemory.close();
		}
		for (MappedMemoryManager mem:this.readOnlyMemory.values()){
			if (mem!=null) mem.close();
		}
	}
	public void deleteMemoryFiles(){
		if(this.writeableMemory != null) {
			this.writeableMemory.delete();
		}
		for (MappedMemoryManager mem:this.readOnlyMemory.values()){
			if (mem!=null) mem.delete();
		}		
	}
	private void populatePersistedMemoryItems() {
//		this.writeableMemory.getMemoryMapBuffer().load();
	}
	private void persistExistingMemoryItems() {
		MemoryList newMemory = new MemoryList();
		boolean found = false;
		for (MemoryItem item:this.currentMemory){
			found = true;
			PersistentMemoryItem pitem = new PersistentMemoryItem(item.getId(),item.getValue(),item.getType(),this);
			newMemory.add(pitem);
		}
		if (found)this.currentMemory = newMemory;
	}
}
